package wang.yang;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Scanner;

/**
 * 负责任务的创建和提交
 * 舰艇任务结果的创建
 * 打印结果
 */
public class MyClient {
    private static final Logger LOG = LoggerFactory.getLogger(MyClient.class);

    ZooKeeper zk;
    String hostPort;
    volatile boolean connected = false;
    volatile boolean expired = false;

    MyClient(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws Exception {
        zk = new ZooKeeper(hostPort, 5000, (e) -> {
            System.out.println(e);

            if (e.getType() == Watcher.Event.EventType.None) {
                switch (e.getState()) {
                    case SyncConnected:
                        connected = true;
                        break;
                    case Disconnected:
                        connected = false;
                        break;
                    case Expired:
                        expired = true;
                        connected = false;
                        LOG.warn("Exiting due to session expiration");
                    default:
                        break;
                }
            }
        });
    }

    // 创建顺序节点
    void submitTask(Task task) {
        zk.create("/unassignedTasks/task-",
                task.getCommand().getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT_SEQUENTIAL,
                createTaskCallback,
                task);
    }

    AsyncCallback.StringCallback createTaskCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int rc, String path, Object task, String nodeName) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    // 可能会重复创建任务
                    submitTask((Task) task);
                    break;
                case OK:
                    LOG.info("My created taskId: {}, command: {}", nodeName, ((Task) task).getCommand());
                    ((Task) task).setTaskId(nodeName);
                    checkResult(nodeName.replace("/unassignedTasks/", "/status/"), (Task) task);
                    break;
                default:
                    LOG.error("Something went wrong" + KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    /**
     * 监控任务状态
     *
     * @param path 状态节点路径
     * @param task 任务上下文
     */
    void checkResult(String path, Task task) {
        // 监听任务状态节点的创建
        zk.exists(path,
                statusWatcher,
                existsCallback,
                task);
    }

    Watcher statusWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent e) {
            if (e.getType() == Event.EventType.NodeCreated) {
                // 获取任务结果
                zk.getData(e.getPath(),
                        false,
                        getTaskResultCallback,
                        null);
            }
        }
    };

    AsyncCallback.StatCallback existsCallback = new AsyncCallback.StatCallback() {
        @Override
        public void processResult(int rc, String path, Object task, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkResult(path, (Task) task);
                    break;
                case OK:
                    if (stat != null) {
                        // 任务状态节点存在，打印结果
                        zk.getData(path, false, getTaskResultCallback, task);
                        LOG.info("获取到任务结果: " + path);
                    }
                    break;
                case NONODE:
                    // 监听状态节点的创建
                    break;
                default:
                    LOG.error("Something went wrong when " +
                            "checking if the status node exists: " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
                    break;
            }
        }
    };

    AsyncCallback.DataCallback getTaskResultCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int rc, String path, Object task, byte[] data, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    /*
                     * Try again.
                     */
                    zk.getData(path, false, getTaskResultCallback, null);
                    return;
                case OK:
                    /*
                     *  Print result
                     */
                    String taskResult = new String(data);
                    LOG.info("Task " + path + ", 执行结果：" + taskResult);
                    /*
                     *  Delete status znode
                     */
                    zk.delete(path, -1, deleteStatusNodeCallback, null);
                    break;
                case NONODE:
                    LOG.warn("Status node is gone!");
                    return;
                default:
                    LOG.error("Something went wrong here, " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    AsyncCallback.VoidCallback deleteStatusNodeCallback = new AsyncCallback.VoidCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    zk.delete(path, -1, deleteStatusNodeCallback, null);
                    break;
                case OK:
                    LOG.info("Successfully deleted " + path);
                    break;
                default:
                    LOG.error("Something went wrong here, " +
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    public static void main(String args[]) throws Exception {
        MyClient myClient = new MyClient("127.0.0.1:2181");
        myClient.startZK();

        while (!myClient.connected) {
            Thread.sleep(100);
        }

        Scanner scanner = new Scanner(System.in);
        while (!myClient.expired) {
            Task task = new Task();
            task.setCommand(scanner.nextLine());
            myClient.submitTask(task);
        }

    }
}